package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: kongcb
 * @Description: TODO
 * @Date: 2021/12/15 9:23
 */
public class TImportExportDetail {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_import_export_detail").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_customs_register");
        long zs_count = zs_data.count();
        if(zs_count>0){

            //企查查进出口表数据取统一社会信用代码
            spark.sql("select a.*,b.credit_code as creditno from a_dc_ep_dwi.t_import_export_detail a left join a_dc_ep_ods.t_eci_company b on a.company_id = b.company_id where b.company_id is not null")
                    .createOrReplaceTempView("qcc_import_export_detail");
            //中数数据表获取相应的有用字段
            spark.sql("SELECT b.*,\n" +
                    "          c.CRECODE,\n" +
                    "          c.CRERATING,\n" +
                    "          c.CREDATE,\n" +
                    "          c.CRERATING_CODE\n" +
                    "   FROM a_dc_ep_ods.zs_t_customs_register b\n" +
                    "   LEFT JOIN a_dc_ep_ods.zs_t_customs_credit c ON b.entid = c.entid\n" +
                    "   WHERE c.entid IS NOT NULL ").createOrReplaceTempView("zs_table1");
            //新增
            spark.sql("insert overwrite table a_dc_ep_incr.t_import_export_detail SELECT " +
                    "       zs.RECORD_ID,\n" +
                    "       qcc.key_no,\n" +
                    "       qcc.company_id,\n" +
                    "       qcc.company_name,\n" +
                    "       zs.GACCID,\n" +
                    "       zs.REGDATE,\n" +
                    "       zs.REGGACC,\n" +
                    "       zs.ADMDIV,\n" +
                    "       zs.APANAGE,\n" +
                    "       if(zs.TRADETYPE='进出口收发货人','进出口货物收发货人',zs.TRADETYPE),\n" +
                    "       zs.SPETXZONE,\n" +
                    "       zs.INDUSTRY,\n" +
                    "       zs.VALIDDATE,\n" +
                    "       '' AS ebusiness_type,\n" +
                    "       zs.STATUS,\n" +
                    "       zs.ANNREPORT,\n" +
                    "       '' AS punish_info,\n" +
                    "       zs.CRERATING_CODE,\n" +
                    "       zs.CRECODE,\n" +
                    "       zs.CRERATING,\n" +
                    "       zs.CREDATE,\n" +
                    "       zs.jobid,\n" +
                    "       '1' AS isadd\n" +
                    "FROM zs_table1 zs\n" +
                    "LEFT JOIN\n" +
                    "  (SELECT a.key_no,\n" +
                    "          a.company_id,\n" +
                    "          a.company_name,\n" +
                    "          a.credit_code\n" +
                    "   FROM a_dc_ep_ods.t_eci_company a\n" +
                    "   WHERE a.credit_code IS NOT NULL\n" +
                    "     AND a.credit_code !='') qcc ON upper(trim(qcc.credit_code)) = upper(trim(zs.CREDIT_CODE))\n" +
                    "WHERE qcc.credit_code IS NOT NULL");
            //新增
            spark.sql("insert into table a_dc_ep_incr.t_import_export_detail SELECT " +
                    "       zs.RECORD_ID,\n" +
                    "       qcc.key_no,\n" +
                    "       qcc.company_id,\n" +
                    "       qcc.company_name,\n" +
                    "       zs.GACCID,\n" +
                    "       zs.REGDATE,\n" +
                    "       zs.REGGACC,\n" +
                    "       zs.ADMDIV,\n" +
                    "       zs.APANAGE,\n" +
                    "       if(zs.TRADETYPE='进出口收发货人','进出口货物收发货人',zs.TRADETYPE),\n" +
                    "       zs.SPETXZONE,\n" +
                    "       zs.INDUSTRY,\n" +
                    "       zs.VALIDDATE,\n" +
                    "       '' AS ebusiness_type,\n" +
                    "       zs.STATUS,\n" +
                    "       zs.ANNREPORT,\n" +
                    "       '' AS punish_info,\n" +
                    "       zs.CRERATING_CODE,\n" +
                    "       zs.CRECODE,\n" +
                    "       zs.CRERATING,\n" +
                    "       zs.CREDATE,\n" +
                    "       zs.jobid,\n" +
                    "       '1' AS isadd\n" +
                    "FROM zs_table1 zs\n" +
                    "LEFT JOIN\n" +
                    "  (SELECT a.key_no,\n" +
                    "          a.company_id,\n" +
                    "          a.company_name,\n" +
                    "          a.credit_code\n" +
                    "   FROM a_dc_ep_ods.t_eci_company a\n" +
                    "   WHERE a.credit_code is NOT NULL\n" +
                    "     AND a.credit_code ='') qcc ON trim(qcc.company_name) = trim(zs.entname)\n" +
                    "WHERE qcc.company_name IS NOT NULL");
            //删除
            spark.sql("insert into table a_dc_ep_incr.t_import_export_detail select " +
                    "qcc.id,\n" +
                    "qcc.key_no,\n" +
                    "qcc.company_id,\n" +
                    "qcc.company_name,\n" +
                    "qcc.reg_no,\n" +
                    "qcc.reg_date,\n" +
                    "qcc.reg_gov,\n" +
                    "qcc.administrative_area,\n" +
                    "qcc.economic_area,\n" +
                    "qcc.trade_type,\n" +
                    "qcc.special_trade_area,\n" +
                    "qcc.industry_type,\n" +
                    "qcc.expire_date,\n" +
                    "qcc.ebusiness_type,\n" +
                    "qcc.cancellation_flag,\n" +
                    "qcc.annual_status,\n" +
                    "qcc.punish_info,\n" +
                    "qcc.credit_no,\n" +
                    "qcc.credit_code,\n" +
                    "qcc.credit_grade,\n" +
                    "qcc.credit_time,\n" +
                    "qcc.dates,\n" +
                    "'-1' as isadd " +
                    "from qcc_import_export_detail qcc inner join (select del.* from a_dc_ep_ods.zs_t_customs_register_del del where del.creditcode is not null and del.creditcode !='') zs  on upper(trim(qcc.creditno)) = upper(trim(zs.creditcode))");

        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_import_export_detail");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }

        spark.stop();
    }
}
